{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n",
      "(402, 3352)\n",
      "(402,)\n"
     ]
    }
   ],
   "source": [
    "#采取嵌套流水线模式\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from jqdata import *\n",
    "from jqfactor import get_factor_values\n",
    "from jqfactor import get_all_factors \n",
    "\n",
    "code='300750.XSHE'#这里以宁德时代股票为例子\n",
    "date_1='2024-01-01'#设置训练数据开始日期\n",
    "date_2='2025-08-31'#设置训练数据结束日期\n",
    "\n",
    "\n",
    "#定义滞后阶数\n",
    "lags = [1, 2, 3]\n",
    "\n",
    "#获取历史股价及计算跨日均值收益率\n",
    "df=get_price(code, start_date=date_1, end_date=date_2, frequency='daily', fields=['close','avg'], skip_paused=False, fq='pre', count=None, panel=False, fill_paused=True)\n",
    "df= df.sort_index() #确保数据df是按时间顺序排序\n",
    "y_0=df['avg']\n",
    "y = np.log(y_0.shift(-1)/y_0) # index=k，则y.iloc[k]是k到 min{t∈T_y|t>k} 日的收益率,后续可看模型效果选择对数收益率\n",
    "\n",
    "X=pd.DataFrame() #创建空dataframe用于存储调仓日前一天获取的输入特征\n",
    "\n",
    "#用调仓日前一天的股票收盘价作为特征，也可换为交易时间点的股价\n",
    "X['price_close']=df['close']\n",
    "\n",
    "#添加交易日前一天的每15min的变化率作为特征\n",
    "df = get_price(code, start_date=date_1, end_date=date_2, frequency='15m',fields=['open','close'])\n",
    "df['date'] = df.index.strftime('%Y-%m-%d')\n",
    "df['time'] = df.index.strftime('%H:%M:%S')\n",
    "df['diff_ratio'] = np.log(df['close']/df['open'])#注意这里取对数了\n",
    "df_1 = df.pivot(index='date', columns='time', values='diff_ratio')\n",
    "X=X.join(df_1, how='left')\n",
    "#添加交易日前一天的每15min的变化率作为特征\n",
    "\n",
    "#添加指数作为特征\n",
    "stock_index = get_all_securities(types=['index'], date=None)\n",
    "for i in stock_index.index:\n",
    "    df = get_price(i, start_date=date_1, end_date=date_2,frequency='daily', fields=['open', 'close', 'volume'])\n",
    "    df= df.sort_index()\n",
    "    df['ratio'] = np.log(df['close']/df['open'])\n",
    "    #同时构建df['ratio']的滞后特征\n",
    "    for lag in lags:\n",
    "        df[f'ratio_{lag}'] = df['ratio'].shift(lag)\n",
    "    df.drop(columns=['open'], axis=1, inplace=True)\n",
    "    df = df.rename(columns=lambda x: f\"{x}_{i}\")\n",
    "    X = X.join(df, how='left')\n",
    "    \n",
    "#添加资金流向作为特征\n",
    "df=get_money_flow(code, start_date=date_1, end_date=date_2, fields=None, count=None)\n",
    "df=df.set_index('date')\n",
    "df=df.drop(columns='sec_code')\n",
    "X = X.join(df, how='left')\n",
    "\n",
    "#添加基本面因子作为特征\n",
    "df = get_all_factors()\n",
    "judge = (df['category'] == 'basics')\n",
    "factors_list = df[judge]['factor'].tolist()\n",
    "factor_data = get_factor_values(securities=code, factors=factors_list, start_date=date_1, end_date=date_2)\n",
    "for key in factor_data.keys():\n",
    "    df_factor=factor_data[key].rename(columns=lambda x: f\"{x}_{key}\")\n",
    "    X = X.join(df_factor, how='left')\n",
    "\n",
    "X = X.shift(1)  \n",
    "# 执行前行标为k的行代表的是第k日获取的输入向量，执行后行为k的代表的是第 max{t∈T_X|t<k}日获取的输入向量\n",
    "\n",
    "X_tail_5=X.tail(5) #将X的最后5行存储到X_tail_5中\n",
    "columns_NotUpdate=X.columns[X_tail_5.isna().all()] #将全为na值的列的列标存储到变量columns_drop中\n",
    "X=X.drop(columns=columns_NotUpdate)\n",
    "\n",
    "#删除因构建滞后特征而出现整行确实的第一行\n",
    "X=X.iloc[1:]\n",
    "y=y.iloc[1:]\n",
    "\n",
    "print(X.index.equals(y.index))  #判断X和y的行标签是否完全一致，在批量训练的时候可以写为if判断条件\n",
    "print(X.shape)\n",
    "print(y.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.35\n"
     ]
    }
   ],
   "source": [
    "#导入所需的库和函数\n",
    "import sklearn\n",
    "import pandas as pd\n",
    "from sklearn.base import BaseEstimator, TransformerMixin\n",
    "from sklearn.pipeline import Pipeline\n",
    "from sklearn.feature_selection import SelectKBest, mutual_info_regression\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "from sklearn.ensemble import RandomForestRegressor\n",
    "from sklearn.model_selection import train_test_split, cross_val_score, TimeSeriesSplit\n",
    "from sklearn.model_selection import GridSearchCV  # 网格搜索\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score\n",
    "from xgboost import XGBRegressor\n",
    "\n",
    "#数据划分\n",
    "X_train, X_test, y_train, y_test = train_test_split(\n",
    "    X, y,\n",
    "    test_size=0.2,  # 测试集比例\n",
    "    shuffle=False   # 关闭随机打乱，保持时间顺序\n",
    ")\n",
    "\n",
    "#y_test的最后一行是缺失值，故舍弃最后一个样本\n",
    "y_test = y_test.iloc[:-1]\n",
    "X_test = X_test.iloc[:-1]\n",
    "\n",
    "#下面四个是自定义的数据预处理步骤\n",
    "class DropNotUpdateFeatures(BaseEstimator, TransformerMixin):\n",
    "    def fit(self, X, y=None):\n",
    "        return self  # 必须返回self\n",
    "    def transform(self, X):\n",
    "        columns_to_drop = X.columns.intersection(columns_NotUpdate)# 1. 找出X中存在且在columns_NotUpdate中的列（即需要删除的列）\n",
    "        X_drop = X.drop(columns=columns_to_drop, errors='ignore')\n",
    "        return X_drop  # DataFrame按列索引筛选\n",
    "\n",
    "\n",
    "class DropHighMissingFeatures(BaseEstimator, TransformerMixin):\n",
    "    def __init__(self, threshold=0.2):\n",
    "        self.threshold = threshold  # 缺失率阈值（20%）\n",
    "        self.keep_cols = None  # 用于保存训练集中需要保留的特征列索引\n",
    "    def fit(self, X, y=None):\n",
    "        # 计算每个特征的缺失值比例（X可以是numpy数组或DataFrame）\n",
    "        missing_ratio=X.isnull().mean(axis=0)\n",
    "        # 筛选出缺失率 <= 阈值的特征列索引（需要保留的列）\n",
    "        self.keep_cols = list(missing_ratio[missing_ratio < self.threshold].index)\n",
    "        return self  # 必须返回self\n",
    "    def transform(self, X):\n",
    "        # 只保留fit阶段确定的特征列\n",
    "        return X.loc[:, self.keep_cols]  # DataFrame按列索引筛选\n",
    "\n",
    "class LinearInterpolator(BaseEstimator, TransformerMixin):\n",
    "    \"\"\"\n",
    "    基于线性插值的缺失值填充转换器。\n",
    "    \n",
    "    核心功能：\n",
    "    - fit_transform：使用双向线性插值（both）填充缺失值\n",
    "    - transform：使用前向线性插值（forward）填充缺失值\n",
    "    \n",
    "    适用于数值型数据的缺失值处理，可集成到scikit-learn流水线中。\n",
    "    \"\"\"\n",
    "    def __init__(self):\n",
    "        # 初始化存储删除行索引的属性\n",
    "        self.deleted_indices_ = None\n",
    "    def fit(self, X, y=None):\n",
    "        \n",
    "        return self\n",
    "    \n",
    "    def transform(self, X):\n",
    "        \"\"\"\n",
    "        使用前向线性插值填充缺失值。\n",
    "        \n",
    "        参数:\n",
    "            X: 类数组 (n_samples, n_features)，包含缺失值的输入数据\n",
    "        \n",
    "        返回:\n",
    "            X_filled: numpy数组 (n_samples, n_features)，填充后的数据集\n",
    "        \"\"\"\n",
    "        # 转换为DataFrame以使用pandas的插值功能\n",
    "        X_df = pd.DataFrame(X)\n",
    "        # 线性插值，仅向前填充（使用前面的有效值）\n",
    "        X_filled = X_df.interpolate(method='linear', limit_direction='forward')\n",
    "        # 记录删除前的所有行索引\n",
    "        original_indices = X_filled.index\n",
    "        # 删除仍含缺失值的行\n",
    "        X_dropna = X_filled.dropna(axis=0)\n",
    "        # 计算被删除的行索引（原始索引 - 保留的索引）\n",
    "        self.deleted_indices_ = original_indices.difference(X_dropna.index)        \n",
    "        return X_dropna\n",
    "    \n",
    "    def fit_transform(self, X, y=None):\n",
    "        \"\"\"\n",
    "        使用双向线性插值填充缺失值（覆盖默认实现）。\n",
    "        \n",
    "        参数:\n",
    "            X: 类数组 (n_samples, n_features)，包含缺失值的输入数据\n",
    "            y: 忽略（仅为保持接口兼容性）\n",
    "        \n",
    "        返回:\n",
    "            X_filled: numpy数组 (n_samples, n_features)，填充后的数据集\n",
    "        \"\"\"\n",
    "        X_df = pd.DataFrame(X)\n",
    "        # 线性插值，双向填充（可使用前后的有效值）\n",
    "        X_filled = X_df.interpolate(method='linear', limit_direction='both')\n",
    "        return X_filled\n",
    "    \n",
    "\n",
    "class CheckAllMissingColumnTransformer(BaseEstimator, TransformerMixin):\n",
    "    \"\"\"\n",
    "    用于检查输入特征集合中是否存在全缺失值列的转换器\n",
    "    \n",
    "    在transform过程中，若检测到任何一列全部为缺失值则报错，否则返回原特征集合\n",
    "    \"\"\"\n",
    "    \n",
    "    def fit(self, X, y=None):\n",
    "        \n",
    "        return self\n",
    "    \n",
    "    def transform(self, X):\n",
    "        \"\"\"\n",
    "        转换方法：检查全缺失值列并返回结果\n",
    "        \n",
    "        参数:\n",
    "            X: 输入特征集合（可以是pandas DataFrame、numpy数组等二维结构）\n",
    "        \n",
    "        返回:\n",
    "            X: 原特征集合（若未检测到全缺失值列）\n",
    "        \n",
    "        异常:\n",
    "            ValueError: 若检测到存在全缺失值列时抛出\n",
    "        \"\"\"\n",
    "        # 将输入转换为DataFrame以便统一处理列操作\n",
    "        X_df = pd.DataFrame(X)\n",
    "        \n",
    "        # 检查每一列是否全部为缺失值\n",
    "        all_missing_columns = X_df.isna().all()\n",
    "        \n",
    "        # 若存在全缺失值列则报错\n",
    "        if all_missing_columns.any():\n",
    "            # 获取全缺失值列的索引/名称\n",
    "            problematic_cols = all_missing_columns[all_missing_columns].index.tolist()\n",
    "            raise ValueError(\n",
    "                f\"输入特征集合中存在全为缺失值的列，列标识为: {problematic_cols}。\"\n",
    "                \"请处理这些列后再进行后续操作。\"\n",
    "            )\n",
    "        \n",
    "        # 无全缺失值列时返回原输入\n",
    "        return X\n",
    "    def fit_transform(self, X, y=None):\n",
    "        return X\n",
    "\n",
    "#上面面四个是自定义的数据预处理步骤\n",
    "\n",
    "#下面是定义互信息选择转换器\n",
    "feature_selector = SelectKBest(\n",
    "    score_func=mutual_info_regression,  # 回归任务的互信息计算\n",
    "    k=80 # 选里取80个特征，后面可以考虑到该步时的输入特征集的行、列数特定比例\n",
    ")\n",
    "\n",
    "\n",
    "complex_preprocessor = Pipeline([\n",
    "    ('dropNotUpdate', DropNotUpdateFeatures()),  # 第一步：丢弃不更新的列\n",
    "    ('drop_high_missing', DropHighMissingFeatures(threshold=0.2)),  # 第二步：删除高缺失特征\n",
    "    ('interpolator', LinearInterpolator()),  # 第三步：填充缺失值\n",
    "    (\"scaler\", StandardScaler()),  # 第四步：标准化（可选，不影响互信息计算）\n",
    "    (\"feature_selector\", feature_selector), #第五步：用互信息选择特征\n",
    "    ('CheckAllMissingColumn', CheckAllMissingColumnTransformer()),  # 第5.5步：预测时若有整列缺失，则中止\n",
    "])\n",
    "\n",
    "\n",
    "#下面是定义xgboost预测器\n",
    "xgb = XGBRegressor(\n",
    "            objective='reg:linear',\n",
    "            random_state=42,\n",
    "        )\n",
    "\n",
    "\n",
    "\n",
    "# 第一阶段：快速找到大致范围\n",
    "param_grid_stage1 = {\n",
    "    'n_estimators': [50, 80],\n",
    "    'max_depth': [4, 6],\n",
    "    'learning_rate': [0.05, 0.1],\n",
    "    'reg_alpha': [ 0.1, 1],\n",
    "    'reg_lambda': [1, 2]\n",
    "}\n",
    "\n",
    "\n",
    "outer_pipeline = Pipeline([\n",
    "    ('preprocessing', complex_preprocessor),\n",
    "    ('model_selection', GridSearchCV(\n",
    "        estimator=xgb,  \n",
    "        param_grid=param_grid_stage1,\n",
    "        cv=5,\n",
    "        refit=True\n",
    "    ))\n",
    "])\n",
    "\n",
    "outer_pipeline.fit(X_train, y_train)\n",
    "y_test_predict=outer_pipeline.predict(X_test)\n",
    "\n",
    "def sign_diff_ratio(arr1, arr2):\n",
    "    \"\"\"\n",
    "    计算两个numpy数组中相同位置正负符号不同的元素占比\n",
    "    \n",
    "    参数:\n",
    "        arr1, arr2: 形状相同的数值型numpy数组\n",
    "    返回:\n",
    "        float: 符号不同的元素占比\n",
    "    \"\"\"\n",
    "    # 检查数组形状是否一致\n",
    "    if arr1.shape != arr2.shape:\n",
    "        raise ValueError(\"两个数组必须具有相同的形状\")\n",
    "    \n",
    "    # 计算各数组元素的正负掩码（True表示满足条件）\n",
    "    arr1_positive = arr1 > 0  # arr1中元素为正的位置\n",
    "    arr1_negative = arr1 < 0  # arr1中元素为负的位置\n",
    "    arr2_positive = arr2 > 0  # arr2中元素为正的位置\n",
    "    arr2_negative = arr2 < 0  # arr2中元素为负的位置\n",
    "    \n",
    "    # 符号不同的掩码：(arr1正且arr2负) 或 (arr1负且arr2正)\n",
    "    diff_mask = (arr1_positive & arr2_negative) | (arr1_negative & arr2_positive)\n",
    "    \n",
    "    # 计算占比：符号不同的元素数 / 总元素数\n",
    "    return np.sum(diff_mask) / arr1.size\n",
    "\n",
    "print(sign_diff_ratio(y_test, y_test_predict))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "MarkDown菜单",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
